#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
隐蔽定时任务权限维持模块单元测试
"""

import unittest
import tempfile
import os
import shutil
import subprocess
from unittest.mock import Mock, patch, MagicMock

# 添加模块路径
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))

from cron_hide.core import CronHidePersistenceModule
from base import PersistenceConfig, PersistenceResult, PersistenceStatus, PersistenceLevel


class TestCronHidePersistenceModule(unittest.TestCase):
    """隐蔽定时任务权限维持模块测试"""
    
    def setUp(self):
        """测试初始化"""
        self.module = CronHidePersistenceModule()
        self.test_config = PersistenceConfig(
            payload="echo 'test payload'",
            stealth_level=7
        )
        
        # 创建临时目录
        self.temp_dir = tempfile.mkdtemp()
        
    def tearDown(self):
        """测试清理"""
        # 清理临时目录
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)
    
    def test_module_initialization(self):
        """测试模块初始化"""
        self.assertIsInstance(self.module, CronHidePersistenceModule)
        self.assertEqual(self.module.name, "cron_hide")
        self.assertIn("linux", self.module.supported_platforms)
    
    def test_config_validation(self):
        """测试配置验证"""
        # 有效配置
        result = self.module.validate_config(self.test_config)
        self.assertTrue(result.success)
        
        # 无效配置 - 空payload
        invalid_config = PersistenceConfig(
            payload="",
            stealth_level=5
        )
        result = self.module.validate_config(invalid_config)
        self.assertFalse(result.success)
        
        # 无效配置 - 无效隐蔽级别
        invalid_config = PersistenceConfig(
            payload="echo test",
            stealth_level=15
        )
        result = self.module.validate_config(invalid_config)
        self.assertFalse(result.success)
    
    def test_platform_support(self):
        """测试平台支持检查"""
        with patch('os.name', 'posix'):
            self.assertTrue(self.module.is_platform_supported())
        
        with patch('os.name', 'nt'):
            self.assertFalse(self.module.is_platform_supported())
    
    @patch('subprocess.run')
    def test_cron_service_check(self, mock_run):
        """测试cron服务检查"""
        # cron服务运行中
        mock_run.return_value.returncode = 0
        self.assertTrue(self.module._check_cron_service_active())
        
        # cron服务未运行
        mock_run.return_value.returncode = 1
        self.assertFalse(self.module._check_cron_service_active())
        
        # 命令执行失败
        mock_run.side_effect = subprocess.CalledProcessError(1, 'systemctl')
        self.assertFalse(self.module._check_cron_service_active())
    
    @patch('os.getuid')
    @patch('os.access')
    def test_privileges_check(self, mock_access, mock_getuid):
        """测试权限检查"""
        # root用户
        mock_getuid.return_value = 0
        self.assertTrue(self.module._check_sufficient_privileges())
        
        # 普通用户，有crontab访问权限
        mock_getuid.return_value = 1000
        mock_access.return_value = True
        self.assertTrue(self.module._check_sufficient_privileges())
        
        # 普通用户，无crontab访问权限
        mock_getuid.return_value = 1000
        mock_access.return_value = False
        self.assertFalse(self.module._check_sufficient_privileges())
    
    def test_deployment_strategy_selection(self):
        """测试部署策略选择"""
        # 低隐蔽级别
        strategy = self.module._select_deployment_strategy(3)
        self.assertIn(strategy, ['direct_cron', 'user_crontab'])
        
        # 中等隐蔽级别
        strategy = self.module._select_deployment_strategy(6)
        self.assertIn(strategy, ['system_cron_d', 'user_crontab'])
        
        # 高隐蔽级别
        strategy = self.module._select_deployment_strategy(9)
        self.assertIn(strategy, ['anacron_delayed', 'system_cron_d'])
    
    def test_task_name_generation(self):
        """测试任务名称生成"""
        # 系统级任务名称
        name = self.module._generate_task_name(5, 'system')
        self.assertTrue(len(name) > 0)
        self.assertNotIn('malicious', name.lower())
        self.assertNotIn('backdoor', name.lower())
        
        # 用户级任务名称
        name = self.module._generate_task_name(8, 'user')
        self.assertTrue(len(name) > 0)
        
        # 高隐蔽级别任务名称
        name = self.module._generate_task_name(10, 'system')
        self.assertTrue(len(name) > 0)
        # 高隐蔽级别应该生成更隐蔽的名称
        self.assertIn(name, [
            'systemd-update-check', 'kernel-security-scan',
            'network-health-monitor', 'disk-cleanup-service'
        ])
    
    def test_stealth_location_selection(self):
        """测试隐蔽位置选择"""
        location = self.module._select_stealth_location(7)
        self.assertTrue(len(location) > 0)
        self.assertTrue(location.startswith('/'))
        
        # 高隐蔽级别应该选择更隐蔽的位置
        high_stealth_location = self.module._select_stealth_location(9)
        self.assertIn(high_stealth_location, [
            '/usr/lib/systemd/system-generators',
            '/usr/share/doc/.hidden',
            '/var/cache/fontconfig'
        ])
    
    def test_schedule_generation(self):
        """测试调度时间生成"""
        schedule = self.module._generate_random_schedule(5)
        self.assertTrue(len(schedule) > 0)
        
        # 验证cron格式
        parts = schedule.split()
        self.assertEqual(len(parts), 5)  # 分 时 日 月 周
        
        # 高隐蔽级别应该生成更随机的调度
        high_stealth_schedule = self.module._generate_random_schedule(9)
        self.assertTrue(len(high_stealth_schedule) > 0)
    
    def test_cron_entry_generation(self):
        """测试cron条目生成"""
        schedule = "0 2 * * *"
        command = "echo test"
        
        entry = self.module._generate_cron_entry(schedule, command, 7)
        self.assertIn(schedule, entry)
        self.assertIn(command, entry)
        
        # 高隐蔽级别应该包含更多伪装
        high_stealth_entry = self.module._generate_cron_entry(schedule, command, 9)
        self.assertIn('>/dev/null 2>&1', high_stealth_entry)
    
    @patch('subprocess.run')
    def test_anacron_deployment(self, mock_run):
        """测试anacron部署"""
        mock_run.return_value.returncode = 0
        
        result = self.module._deploy_anacron_delayed(
            "test_task", "echo test", 7
        )
        
        self.assertTrue(result.success)
        self.assertIsNotNone(result.instance_id)
    
    @patch('builtins.open', create=True)
    @patch('os.path.exists')
    def test_system_cron_deployment(self, mock_exists, mock_open):
        """测试系统cron.d部署"""
        mock_exists.return_value = True
        mock_file = MagicMock()
        mock_open.return_value.__enter__.return_value = mock_file
        
        result = self.module._deploy_system_cron_d(
            "test_task", "echo test", 7
        )
        
        self.assertTrue(result.success)
        self.assertIsNotNone(result.instance_id)
    
    @patch('subprocess.run')
    def test_user_crontab_deployment(self, mock_run):
        """测试用户crontab部署"""
        # 模拟crontab -l输出
        mock_run.return_value.stdout = "# existing cron jobs\n"
        mock_run.return_value.returncode = 0
        
        result = self.module._deploy_user_crontab(
            "test_task", "echo test", 7
        )
        
        self.assertTrue(result.success)
        self.assertIsNotNone(result.instance_id)
    
    @patch('subprocess.run')
    def test_direct_cron_deployment(self, mock_run):
        """测试直接cron部署"""
        mock_run.return_value.returncode = 0
        
        result = self.module._deploy_direct_cron(
            "test_task", "echo test", 7
        )
        
        self.assertTrue(result.success)
        self.assertIsNotNone(result.instance_id)
    
    def test_module_info(self):
        """测试模块信息获取"""
        info = self.module.get_module_info()
        
        self.assertEqual(info['name'], 'cron_hide')
        self.assertIn('version', info)
        self.assertIn('description', info)
        self.assertIn('supported_platforms', info)
        self.assertIn('linux', info['supported_platforms'])
    
    @patch.object(CronHidePersistenceModule, '_check_cron_service_active')
    @patch.object(CronHidePersistenceModule, '_check_sufficient_privileges')
    @patch.object(CronHidePersistenceModule, '_deploy_system_cron_d')
    def test_deploy_flow(self, mock_deploy, mock_privileges, mock_cron):
        """测试部署流程"""
        # 设置mock返回值
        mock_cron.return_value = True
        mock_privileges.return_value = True
        mock_deploy.return_value = PersistenceResult(
            success=True,
            message="部署成功",
            instance_id="test_instance"
        )
        
        # 执行部署
        result = self.module.deploy(self.test_config)
        
        self.assertTrue(result.success)
        self.assertIsNotNone(result.instance_id)
    
    def test_instance_management(self):
        """测试实例管理"""
        # 测试保存实例配置
        instance_id = "test_instance"
        config_data = {
            'task_name': 'test_task',
            'strategy': 'system_cron_d',
            'stealth_level': 7
        }
        
        self.module._save_instance_config(instance_id, config_data)
        
        # 测试加载实例配置
        loaded_config = self.module._load_instance_config(instance_id)
        self.assertEqual(loaded_config['task_name'], 'test_task')
        
        # 测试删除实例配置
        self.module._delete_instance_config(instance_id)
        deleted_config = self.module._load_instance_config(instance_id)
        self.assertIsNone(deleted_config)


class TestCronHideModuleIntegration(unittest.TestCase):
    """隐蔽定时任务模块集成测试"""
    
    def setUp(self):
        """集成测试初始化"""
        self.module = CronHidePersistenceModule()
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_crontab_availability(self):
        """测试crontab命令可用性"""
        try:
            result = subprocess.run(['crontab', '-l'], 
                                  capture_output=True, text=True)
            # crontab -l 可能返回非0状态码（如果没有crontab），但命令应该存在
            self.assertTrue(True)  # 如果能执行到这里说明命令存在
        except FileNotFoundError:
            self.fail("crontab命令不可用")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_cron_directories_exist(self):
        """测试cron相关目录存在性"""
        # 检查常见的cron目录
        cron_dirs = [
            '/etc/cron.d',
            '/var/spool/cron',
            '/etc/crontab'
        ]
        
        for cron_path in cron_dirs:
            if os.path.exists(cron_path):
                self.assertTrue(True)
                return
        
        # 如果所有目录都不存在，可能是特殊的系统配置
        self.skipTest("未找到标准cron目录结构")
    
    @unittest.skipUnless(os.name == 'posix', "需要Linux环境")
    def test_anacron_availability(self):
        """测试anacron可用性"""
        try:
            subprocess.run(['anacron', '-T'], capture_output=True)
            self.assertTrue(True)
        except FileNotFoundError:
            self.skipTest("anacron不可用")
    
    def test_schedule_parsing(self):
        """测试调度时间解析"""
        # 测试各种cron格式
        valid_schedules = [
            "0 2 * * *",      # 每天2点
            "*/5 * * * *",    # 每5分钟
            "0 0 1 * *",      # 每月1号
            "0 0 * * 0",      # 每周日
            "@daily",         # 每日
            "@hourly",        # 每小时
        ]
        
        for schedule in valid_schedules:
            # 这里可以添加调度时间验证逻辑
            self.assertTrue(len(schedule) > 0)
    
    def test_command_escaping(self):
        """测试命令转义"""
        dangerous_commands = [
            "echo 'test'; rm -rf /",
            "$(whoami)",
            "`id`",
            "test && malicious_command"
        ]
        
        for cmd in dangerous_commands:
            # 测试命令是否被正确转义
            escaped = self.module._escape_command(cmd)
            self.assertNotEqual(cmd, escaped)
            # 确保危险字符被转义
            self.assertNotIn('$(', escaped)
            self.assertNotIn('`', escaped)


def run_all_tests():
    """运行所有测试"""
    # 创建测试套件
    test_suite = unittest.TestSuite()
    
    # 添加单元测试
    test_suite.addTest(unittest.makeSuite(TestCronHidePersistenceModule))
    
    # 添加集成测试
    test_suite.addTest(unittest.makeSuite(TestCronHideModuleIntegration))
    
    # 运行测试
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(test_suite)
    
    return result.wasSuccessful()


if __name__ == '__main__':
    print("=== 隐蔽定时任务权限维持模块测试 ===")
    
    # 运行所有测试
    success = run_all_tests()
    
    if success:
        print("\n✓ 所有测试通过")
    else:
        print("\n✗ 部分测试失败")
        exit(1)